package com.esunny.dns.ai.task;

import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * 指定一系列任务，内部线程将不断循环检查任务canExecute调件，如果满足条件，执行任务
 * 
 * @author jenvin 2012-6-29 下午3:36:32
 */
public class ConditionTaskCircleQueueImpl implements TaskQueue, Runnable {

    private static final Log      log        = LogFactory.getLog(ConditionTaskCircleQueueImpl.class);
    private int                   workerSize = 0;
    private Thread[]              workers    = null;
    transient final ReentrantLock lock       = new ReentrantLock();
    private ConditionTask[]       taskArray  = new ConditionTask[0];
    private Long                  offerSet   = 0L;
    private volatile boolean      isExit     = false;

    public ConditionTaskCircleQueueImpl(int worker){
        if (worker <= 0) throw new IllegalArgumentException("worker");
        this.workerSize = worker;
    }

    @Override
    public void start() {
        workers = new Thread[workerSize];
        for (int i = 0; i < workers.length; i++) {
            (workers[i] = new Thread(this)).start();
            workers[i].setName(String.valueOf(i));
        }
    }

    /*
     * (non-Javadoc)
     * @see com.esunny.dns.ai.task.Queue#enqueueTask(com.esunny.dns.ai.task.Task)
     */
    @Override
    public void enqueueTask(Task task) {
        if (!(task instanceof ConditionTask)) throw new IllegalArgumentException("task must be ConditionTask");
        ConditionTask[] tasks = new ConditionTask[1];
        tasks[0] = (ConditionTask) task;
        this.enqueueTask(tasks);
    }

    public void enqueueTask(ConditionTask[] tasks) {
        if (tasks == null) return;
        if (tasks.length == 0) return;

        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int len = taskArray.length;
            ConditionTask[] newElements = Arrays.copyOf(taskArray, len + tasks.length);
            System.arraycopy(tasks, 0, newElements, len, tasks.length);
            taskArray = newElements;
            if (log.isInfoEnabled()) {
                log.info(String.format("successfully added  %s task ,queue size: %s", tasks.length, newElements.length));
            }
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void stop() {
        abort();
    }

    @Override
    public void abort() {
        if (log.isInfoEnabled()) {
            log.info("starting abort task queue,queue size:" + taskArray.length);
        }
        isExit = true;
        for (Thread t : workers) {
            try {
                t.interrupt();// 发出中断信号，退出组塞状态
            } catch (Exception ex) {
                log.error(ex);
            }
        }
        for (Thread t : workers) {
            try {
                t.join();
            } catch (Exception ex) {
                log.error(ex);
            }
        }
        if (log.isInfoEnabled()) {
            log.info("successfully abort");
        }
    }

    @Override
    public void run() {
        while (!isExit) {
            int index = 0;
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                index = (int) (offerSet++ % taskArray.length);
            } finally {
                lock.unlock();
            }
            ConditionTask task = taskArray[index];
            if (!task.canExecute()) {
                sleep(100);
                continue;
            }
            try {
                task.execute();
            } catch (Exception ex) {
                log.error(ex);
            }
        }
        if (log.isInfoEnabled()) {
            log.info("worker thread " + Thread.currentThread().getName() + " exit!");
        }
    }

    private void sleep(int time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            log.error(e);
        }
    }

    public static void main(String[] args) {
        // ScheduledExecutorService scheduExe = Executors.newScheduledThreadPool(10);
        ConditionTask[] tasks = new ConditionTask[2];
        tasks[0] = new TimeTask(1, "task1");
        tasks[1] = new TimeTask(1, "task2");

        ConditionTaskCircleQueueImpl impl = new ConditionTaskCircleQueueImpl(1);
        impl.enqueueTask(tasks);
        impl.start();
        ConditionTask[] eTasks = new ConditionTask[2];
        eTasks[0] = new TimeTask(1, "task3");
        eTasks[1] = new TimeTask(1, "task4");
        impl.enqueueTask(eTasks);
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        // impl.abort();

    }

    public static class TimeTask extends ConditionTask {

        private int    i             = 0;
        private String name          = null;
        private Long   lastExeTimeMs = null;

        public TimeTask(int i, String name){
            this.i = i;
            this.name = name;
        }

        @Override
        public void execute() {
            lastExeTimeMs = System.currentTimeMillis();
            System.out.println(name + "-" + i++ + lastExeTimeMs);
        }

        @Override
        public boolean canExecute() {
            if (lastExeTimeMs == null) return true;
            if (System.currentTimeMillis() - lastExeTimeMs >= 1000 * 10) {
                return true;
            } else return false;

        }
    }
}
